package com.atguigu.flink.chapter10.query;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/6/29 09:09
 */
public class QueryWindowDemo {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        //        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        
        // 在 ddl 中定义时间属性
        tEnv.executeSql("create table sensor(" +
                            " id string, " +
                            " ts bigint, " +
                            " vc int, " +
                            " et as to_timestamp_ltz(ts, 3)," +
                            " watermark for et as et - interval '1' second " +
                            ")with(" +
                            "  'connector' = 'filesystem', " +
                            "  'path' = 'input/sensor.txt', " +
                            "  'format' = 'csv' " +
                            ")");
        
        
        // 滚动窗口
        /*tEnv.sqlQuery("select " +
                          "id, " +
                          "tumble_start(et, interval '5' second) stt, " +
                          "tumble_end(et, interval '5' second) edt," +
                          "sum(vc) vc_sum  " +
                          "from sensor " +
                          "group by  id, tumble(et, interval '5' second)")
            .execute()
            .print();*/
    
    
        /*tEnv.sqlQuery("select " +
                          "id, " +
                          "hop_start(et, interval '2' second, interval '5' second) stt, " +
                          "hop_end(et, interval '2' second, interval '5' second) edt, " +
                          "sum(vc) vc_sum  " +
                          "from sensor " +
                          "group by  id, hop(et, interval '2' second, interval '5' second)")
            .execute()
            .print();*/
        
        
        tEnv.sqlQuery("select " +
                          "id, " +
                          "session_start(et, interval '2' second) stt, " +
                          "session_end(et, interval '2' second) edt, " +
                          "sum(vc) vc_sum  " +
                          "from sensor " +
                          "group by  id, session(et, interval '2' second)")
            .execute()
            .print();
        
        
    }
}
/*
grouping window


 */